package yarndeploylogic

import (
	"context"
	"errors"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"strings"
	"sync"
	"time"
	"yunzhan/common/models"
	utils "yunzhan/common/utils"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"
)

type DeployYarnHALogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewDeployYarnHALogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployYarnHALogic {
	return &DeployYarnHALogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// DeployYarnHA HA部署
func (l *DeployYarnHALogic) DeployYarnHA(in *agent.DeployHadoopRequest) (*agent.DeployResponse, error) {
	if in.NodeInfo == nil {
		return &agent.DeployResponse{
			Code:    500,
			Message: "未配置节点信息, 无法执行Yarn集群部署",
		}, errors.New("未配置节点信息, 无法执行Yarn集群部署")
	}

	resp := &agent.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("请求已接收，正在部署 %s 到节点 %s", in.ConfigInfo.ComponentName, in.NodeInfo.Host),
	}

	go l.startDeployment(in)

	return resp, nil
}
func (l *DeployYarnHALogic) startDeployment(in *agent.DeployHadoopRequest) {

	var wg sync.WaitGroup

	wg.Add(1)

	go func() {
		defer wg.Done()
		// 创建 LogManager 实例
		logManager := models.NewLogManager(100, fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port))
		defer logManager.Close()

		//flushInterval := 5 * time.Second // 刷新间隔
		//maxBatchSize := 100              // 最大批量大小
		//retryDelay := 2 * time.Second    // 重试延迟
		//maxRetries := 3                  // 最大重试次数

		//// 创建 ComponentInfoManager 实例
		//componentInfoManager := models.NewComponentInfoManager(
		//	1000,
		//	fmt.Sprintf("http://%s:%d", l.svcCtx.Config.RestServices["apiServer"].RestConf.Host, l.svcCtx.Config.RestServices["apiServer"].RestConf.Port),
		//	flushInterval,
		//	maxBatchSize,
		//	retryDelay,
		//	maxRetries)
		//defer componentInfoManager.Close()
		packageDirectory := in.ConfigInfo.PackagePath
		rtrimDir := utils.Rtrim(packageDirectory, "/")
		yarnVersion := in.ConfigInfo.ComponentName
		installPath := utils.Rtrim(in.ConfigInfo.InstallPath, "/")
		baseInfo := fmt.Sprintf("安装包存放目录: %s, Yarn version: %s, Yarn安装目录: %s", rtrimDir, yarnVersion, installPath)
		step := 1
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", baseInfo, "Yarn", "Server")

		javaHome := "/usr/local/java"
		packageName := utils.Rtrim(yarnVersion, ".tar.gz")
		yarnHome := fmt.Sprintf("%s/%s", installPath, packageName)

		// 检测安装路径是否存在
		checkPathCommand := fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
`, installPath, installPath)

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "执行组件安装路径检查...", "Yarn", "Server")
		output, err := utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件安装路径异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件安装路径完成", "Yarn", "Server")

		if !strings.Contains(output, packageName) {
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包...", "Yarn", "Server")
			tarCommand := fmt.Sprintf("sudo tar -xzf %s/%s -C %s", rtrimDir, yarnVersion, installPath)
			output, err = utils.ExecCommand(tarCommand)
			if err != nil {
				message := fmt.Sprintf("解压安装包异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "解压安装包完成", "Yarn", "Server")
		} else {
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "Yarn安装目录已存在，跳过解压", "Yarn", "Server")
		}

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组...", "Yarn", "Server")
		chownCommand := fmt.Sprintf("sudo chown -R %s:%s %s", in.NodeInfo.Username, in.NodeInfo.Username, yarnHome)
		output, err = utils.ExecCommand(chownCommand)
		if err != nil {
			message := fmt.Sprintf("修改组件所属用户组异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "修改组件所属用户组完成", "Yarn", "Server")

		// 检查数据存储目录是否存在并设置权限
		checkPathCommand = fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
sudo chown -R %s:%s %s
`, in.HadoopConfig.CoreSiteConfig.HadoopTmpDir,
			in.HadoopConfig.CoreSiteConfig.HadoopTmpDir,
			in.NodeInfo.Username,
			in.NodeInfo.Username,
			in.HadoopConfig.CoreSiteConfig.HadoopTmpDir)
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录...", "Yarn", "Server")
		output, err = utils.ExecCommand(checkPathCommand)
		if err != nil {
			message := fmt.Sprintf("检查组件数据存储目录异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "检查组件数据存储目录完成", "Yarn", "Server")
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "加载yarn-site.xml配置文件...", "Yarn", "Server")
		yarnSitePath := fmt.Sprintf("%s/etc/hadoop/yarn-site.xml", packageName)
		yarnConfig, err := utils.LoadConfiguration(yarnSitePath)
		if err != nil {
			message := fmt.Sprintf("加载 yarn-site.xml 配置文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}

		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "加载yarn-site.xml配置文件完成", "Yarn", "Server")

		yarnConfig.SetProperty("yarn.nodemanager.log-dirs", in.HadoopConfig.YarnSiteConfig.YarnNodemanagerLogDirs, "NodeManager日志目录")
		yarnConfig.SetProperty("yarn.nodemanager.delete.debug-delay-sec", in.HadoopConfig.YarnSiteConfig.YarnNodemanagerDeleteDebugDelaySec, "NodeManager删除调试延迟时间（秒）")
		yarnConfig.SetProperty("yarn.log.server.url", in.HadoopConfig.YarnSiteConfig.YarnLogServerUrl, "YARN日志服务器URL")
		yarnConfig.SetProperty("yarn.resourcemanager.scheduler.class", in.HadoopConfig.YarnSiteConfig.YarnResourcemanagerSchedulerClass, "ResourceManager调度器类")
		yarnConfig.SetProperty("yarn.scheduler.fair.preemption", in.HadoopConfig.YarnSiteConfig.YarnSchedulerFairPreemption, "是否启用公平抢占")
		yarnConfig.SetProperty("yarn.log-aggregation.retain-seconds", in.HadoopConfig.YarnSiteConfig.YarnLogAggregationRetainSeconds, "日志聚合保留时间（秒）")
		yarnConfig.SetProperty("yarn.nodemanager.remote-app-log-dir", in.HadoopConfig.YarnSiteConfig.YarnNodemanagerRemoteAppLogDir, "NodeManager远程应用日志目录")
		yarnConfig.SetProperty("yarn.nodemanager.vmem-check-enabled", in.HadoopConfig.YarnSiteConfig.YarnNodemanagerVmemCheckEnabled, "是否启用虚拟内存检查")
		yarnConfig.SetProperty("yarn.log-aggregation-enable", in.HadoopConfig.YarnSiteConfig.YarnLogAggregationEnable, "是否启用日志聚合")
		yarnConfig.SetProperty("yarn.nodemanager.aux-services", in.HadoopConfig.YarnSiteConfig.YarnNodemanagerAuxServices, "NodeManager辅助服务")
		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "更新 yarn-site.xml 配置文件...", "Yarn", "Server")
		err = yarnConfig.Save(yarnSitePath)
		if err != nil {
			message := fmt.Sprintf("更新 yarn-site.xml 配置文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "更新 yarn-site.xml 配置文件完成", "Yarn", "Server")

		step++
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置 yarn-env.sh 文件...", "Yarn", "Server")
		hadoopEnvPath := fmt.Sprintf("%s/etc/hadoop/yarn-env.sh", packageName)
		appendHadoopEnvCommand := fmt.Sprintf(`
sudo bash -c 'cat <<EOF >>%s
export JAVA_HOME=%s
export YARN_RESOURCEMANAGER_OPTS="-XX:+AlwaysPreTouch -Xss1m  -Xms2g -Xmx2g -XX:+UseG1GC -XX:+HeapDumpOnOutOfMemoryError -XX:G1HeapRegionSize=4m -XX:InitiatingHeapOccupancyPercent=30 -XX:G1ReservePercent=15"
EOF'
`, hadoopEnvPath, javaHome)
		output, err = utils.ExecCommand(appendHadoopEnvCommand)
		if err != nil {
			message := fmt.Sprintf("配置 yarn-env.sh 文件异常: %v, output: %s", err, output)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "配置yarn-env.sh文件完成", "Yarn", "Server")

		step++
		currentIP, currentIPErr := utils.GetCurrentInternalIP()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机IP: %v", in.NodeInfo.Host, currentIPErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}
		currentHost, currentHostErr := utils.GetCurrentHostname()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机名: %v", in.NodeInfo.Host, currentHostErr)
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
			return
		}

		if utils.Contains(in.HadoopConfig.ResourcemanagerList, currentHost) || utils.Contains(in.HadoopConfig.ResourcemanagerList, currentIP) {

			resourceServiceFile := fmt.Sprintf(`[Unit]
Description=Hadoop ResourceManager service
After=network.target

[Service]
User=%s
Group=%s
ExecStart=%s/bin/yarn --daemon start resourcemanager
ExecStop=%s/bin/yarn --daemon stop resourcemanager
ExecReload=/bin/kill -HUP $MAINPID
Environment="JAVA_HOME=%s"
Restart=on-failure
RestartSec=10
SuccessExitStatus=0 143
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, yarnHome, yarnHome, javaHome)

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hadoop-resourcemanager.service文件...", "Yarn", "Resourcemanager")
			writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/hadoop-resourcemanager.service", resourceServiceFile)
			_, err = utils.ExecCommand(writeServiceFileCommand)
			if err != nil {
				message := fmt.Sprintf("创建hadoop-resourcemanager.service文件异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Resourcemanager")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hadoop-resourcemanager.service文件完成", "Yarn", "Resourcemanager")

			startCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable hadoop-resourcemanager.service --now")
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Resourcemanager服务...", "Yarn", "Resourcemanager")
			output, err = utils.ExecCommand(startCommand)
			if err != nil {
				message := fmt.Sprintf("启动Resourcemanager服务异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Resourcemanager")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Resourcemanager服务完成", "Yarn", "Resourcemanager")
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待Resourcemanager服务完全启动", "Yarn", "Resourcemanager")

			time.Sleep(3 * time.Second)
			step++
			statusCommand := fmt.Sprintf("sudo systemctl is-active hadoop-resourcemanager.service")
			output, err = utils.ExecCommand(statusCommand)
			if err != nil || strings.TrimSpace(output) != "active" {
				message := fmt.Sprintf("Resourcemanager服务启动后未正常运行: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "Server")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "Resourcemanager服务部署成功", "Yarn", "Resourcemanager")

			yarnInfo := &models.ComponentInfo{
				ClusterID:      in.ClusterID,
				ComponentName:  "Yarn",
				Version:        in.Version,
				NodeHost:       currentHost,
				NodeIP:         currentIP,
				ComponentRole:  "Resourcemanager",
				HomePath:       yarnHome,
				DataStorageDir: "",
				Port:           in.HadoopConfig.YarnSiteConfig.YarnLogServerUrl,
				Status:         true,
				AdditionalInfo: fmt.Sprintf("config_file=%s/etc/hadoop/yarn-site.xml", yarnHome),
			}

			l.svcCtx.ComponentInfoManager.SendComponentInfo(yarnInfo)

		}

		if utils.Contains(in.HadoopConfig.NodemanagerList, currentHost) || utils.Contains(in.HadoopConfig.NodemanagerList, currentIP) {

			nodemanagerServiceFile := fmt.Sprintf(`[Unit]
Description=Hadoop NodeManager service
After=network.target

[Service]
User=%s
Group=%s
ExecStart=%s/bin/yarn --daemon start nodemanager
ExecStop=%s/bin/yarn --daemon stop nodemanager
ExecReload=/bin/kill -HUP $MAINPID
Environment="JAVA_HOME=%s"
Restart=on-failure
RestartSec=10
SuccessExitStatus=0 143
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, yarnHome, yarnHome, javaHome)
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 hadoop-nodemanager.service 文件...", "Yarn", "NodeManager")
			writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/hadoop-nodemanager.service", nodemanagerServiceFile)
			_, err = utils.ExecCommand(writeServiceFileCommand)
			if err != nil {
				message := fmt.Sprintf("创建 hadoop-nodemanager.service 文件异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "NodeManager")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hadoop-nodemanager.service文件完成", "Yarn", "NodeManager")

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Nodemanager服务...", "Yarn", "NodeManager")
			startCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable hadoop-nodemanager.service --now")
			output, err = utils.ExecCommand(startCommand)
			if err != nil {
				message := fmt.Sprintf("启动Nodemanager服务异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "NodeManager")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动Nodemanager服务完成", "Yarn", "NodeManager")
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待Nodemanager服务完全启动", "Yarn", "NodeManager")

			time.Sleep(3 * time.Second)
			step++
			statusCommand := fmt.Sprintf("sudo systemctl is-active hadoop-nodemanager.service")
			output, err = utils.ExecCommand(statusCommand)
			if err != nil || strings.TrimSpace(output) != "active" {
				message := fmt.Sprintf("Nodemanager服务启动后未正常运行: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "NodeManager")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "Nodemanager服务部署成功", "Yarn", "NodeManager")

			yarnInfo := &models.ComponentInfo{
				ClusterID:      in.ClusterID,
				ComponentName:  "Yarn",
				Version:        in.Version,
				NodeHost:       currentHost,
				NodeIP:         currentIP,
				ComponentRole:  "Nodemanager",
				HomePath:       yarnHome,
				DataStorageDir: in.HadoopConfig.YarnSiteConfig.YarnNodemanagerLogDirs,
				Port:           in.HadoopConfig.YarnSiteConfig.YarnLogServerUrl,
				Status:         true,
				AdditionalInfo: fmt.Sprintf("config_file=%s/etc/hadoop/yarn-site.xml", yarnHome),
			}

			l.svcCtx.ComponentInfoManager.SendComponentInfo(yarnInfo)

		}

		if utils.Contains(in.HadoopConfig.NodemanagerList, currentHost) || utils.Contains(in.HadoopConfig.NodemanagerList, currentIP) {
			historyServiceFile := fmt.Sprintf(`[Unit]
Description=Hadoop JobHistoryServer service
After=network.target

[Service]
User=%s
Group=%s
ExecStart=%s/bin/mapred --daemon start historyserver
ExecStop=%s/bin/mapred --daemon stop historyserver
ExecReload=/bin/kill -HUP $MAINPID
Environment="JAVA_HOME=%s"
Restart=on-failure
RestartSec=10
SuccessExitStatus=0 143
LimitNOFILE=65536

[Install]
WantedBy=multi-user.target`, in.NodeInfo.Username, in.NodeInfo.Username, yarnHome, yarnHome, javaHome)
			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建 hadoop-historyserver.service 文件...", "Yarn", "JobHistoryServer")

			writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/hadoop-historyserver.service", historyServiceFile)
			_, err = utils.ExecCommand(writeServiceFileCommand)
			if err != nil {
				message := fmt.Sprintf("创建 hadoop-historyserver.service 文件异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "JobHistoryServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "创建hadoop-historyserver.service文件完成", "Yarn", "JobHistoryServer")

			step++
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动JobHistoryServer服务...", "Yarn", "JobHistoryServer")
			startCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable hadoop-historyserver.service --now")
			output, err = utils.ExecCommand(startCommand)
			if err != nil {
				message := fmt.Sprintf("启动JobHistoryServer服务异常: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "JobHistoryServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "启动JobHistoryServer服务完成", "Yarn", "JobHistoryServer")
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "等待JobHistoryServer服务完全启动", "Yarn", "JobHistoryServer")
			time.Sleep(3 * time.Second)
			step++
			statusCommand := fmt.Sprintf("sudo systemctl is-active hadoop-historyserver.service")
			output, err = utils.ExecCommand(statusCommand)
			if err != nil || strings.TrimSpace(output) != "active" {
				message := fmt.Sprintf("JobHistoryServer服务启动后未正常运行: %v, output: %s", err, output)
				logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "ERROR", message, "Yarn", "JobHistoryServer")
				return
			}
			logManager.SendLog(in.ClusterID, in.NodeInfo.Host, step, "INFO", "JobHistoryServer服务部署成功", "Yarn", "JobHistoryServer")

			yarnInfo := &models.ComponentInfo{
				ClusterID:      in.ClusterID,
				ComponentName:  "Yarn",
				Version:        in.Version,
				NodeHost:       currentHost,
				NodeIP:         currentIP,
				ComponentRole:  "JobHistoryServer",
				HomePath:       yarnHome,
				DataStorageDir: "",
				Port:           in.HadoopConfig.YarnSiteConfig.YarnLogServerUrl,
				Status:         true,
				AdditionalInfo: fmt.Sprintf("config_file=%s/etc/hadoop/yarn-site.xml", yarnHome),
			}

			l.svcCtx.ComponentInfoManager.SendComponentInfo(yarnInfo)
		}

	}()
	wg.Wait()

}
